﻿/*
 * 作者：新生命开发团队（http://www.newlifex.com/）
 * 
 * 版权：版权所有 (C) 新生命开发团队 2002-2014
 * 
 * 修改：海洋饼干（cuteant@outlook.com）
*/

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using CuteAnt.Collections;
using CuteAnt.Configuration;
using CuteAnt.Log;
using ThreadState = System.Threading.ThreadState;

namespace CuteAnt.Threading
{
	/// <summary>线程池。所有静态方法和实例方法均是线程安全。</summary>
	public sealed class ThreadPoolX : IDisposable
	{
		#region -- 基本属性 --

		private Int32 _MaxThreads;

		/// <summary>最大线程数</summary>
		public Int32 MaxThreads
		{
			get { return _MaxThreads; }
			set { _MaxThreads = value; }
		}

		private Int32 _MinThreads;

		/// <summary>最小线程数</summary>
		public Int32 MinThreads
		{
			get { return _MinThreads; }
			set { _MinThreads = value; }
		}

		private String _Name;

		/// <summary>线程池名称</summary>
		public String Name
		{
			get { return _Name; }
			set { _Name = value; }
		}

		private Exception _LastError;

		/// <summary>最后的异常</summary>
		public Exception LastError
		{
			get { return _LastError; }
			set { _LastError = value; }
		}

		#endregion

		#region -- 线程 --

		/// <summary>用于维护管理线程的锁</summary>
		private Object SynLock_mt = new Object();

		/// <summary>使用volatile关键字，等到对象创建完成</summary>
		private volatile Thread _ManagerThread;

		/// <summary>维护线程</summary>
		private Thread ManagerThread
		{
			get
			{
				if (_ManagerThread == null)
				{
					lock (SynLock_mt)
					{
						if (_ManagerThread == null)
						{
							var thread = new Thread(Work);

							//thread.Name = Name + "线程池维护线程";
							thread.Name = Name + "PM";
							thread.IsBackground = true;
							thread.Priority = ThreadPriority.Highest;//最高优先级

							//留到第一个任务到来时再开始维护线程
							//thread.Start();
							_ManagerThread = thread;
						}
					}
				}
				return _ManagerThread;
			}
		}

		/// <summary>第一个任务到来时初始化线程池</summary>
		private void Init()
		{
			if (ManagerThread.IsAlive) { return; }
			if ((ManagerThread.ThreadState & ThreadState.Unstarted) != ThreadState.Unstarted) { return; }
			ManagerThread.Start();
			WriteLog("初始化线程池：" + Name + " 最大：" + MaxThreads + " 最小：" + MinThreads);
		}

		private List<ThreadX> _Threads;

		/// <summary>线程组。适用该资源时，记得加上线程锁lockObj</summary>
		private List<ThreadX> Threads
		{
			get
			{
				if (_Threads == null)
				{
					_Threads = new List<ThreadX>();
				}
				return _Threads;
			}
		}

		private Int32 _ThreadCount;

		/// <summary>当前线程数</summary>
		public Int32 ThreadCount
		{
			get { return _ThreadCount; }
			private set { _ThreadCount = value; }
		}

		private Int32 _RunningCount;

		/// <summary>正在处理任务的线程数</summary>
		public Int32 RunningCount
		{
			get { return _RunningCount; }
			private set { _RunningCount = value; }
		}

		private AutoResetEvent _Event = new AutoResetEvent(false);

		/// <summary>事件量</summary>
		private AutoResetEvent Event
		{
			get { return _Event; }
		}

		/// <summary>用户维护线程组的锁</summary>
		private Object SyncLock_Threads = new object();

		#endregion

		#region -- 任务队列 --

		private SortedList<Int32, ThreadTask> _Tasks;

		/// <summary>任务队列</summary>
		private SortedList<Int32, ThreadTask> Tasks
		{
			get
			{
				if (_Tasks == null)
				{
					_Tasks = new SortedList<Int32, ThreadTask>();
				}
				return _Tasks;
			}
		}

		/// <summary>任务队列同步锁</summary>
		private Object Sync_Tasks = new object();

		#endregion

		#region -- 构造 --

		/// <summary>构造一个线程池</summary>
		/// <param name="name">线程池名</param>
		private ThreadPoolX(String name)
		{
			Name = name;

			//最大线程数为4×处理器个数
			MaxThreads = 10 * Environment.ProcessorCount;
			MinThreads = 2 * Environment.ProcessorCount;
		}

		private static DictionaryCache<String, ThreadPoolX> _cache = new DictionaryCache<String, ThreadPoolX>();

		/// <summary>创建线程池。一个名字只能创建一个线程池。线程安全。</summary>
		/// <param name="name">线程池名</param>
		/// <returns></returns>
		public static ThreadPoolX Create(String name)
		{
			if (name.IsNullOrWhiteSpace())
			{
				throw new ArgumentNullException(name, "线程池名字不能为空！");
			}
			return _cache.GetItem(name, key => new ThreadPoolX(key));
		}

		private static ThreadPoolX _Instance;

		/// <summary>默认线程池</summary>
		public static ThreadPoolX Instance
		{
			get
			{
				if (_Instance == null)
				{
					_Instance = Create("X");
				}
				return _Instance;
			}
			set { _Instance = value; }
		}

		#endregion

		#region -- 队列操作 --

		/// <summary>把用户工作项放入队列</summary>
		/// <param name="method">任务方法</param>
		/// <returns>任务编号</returns>
		public Int32 Queue(WaitCallback method)
		{
			return Queue(method, null);
		}

		/// <summary>把用户工作项放入队列</summary>
		/// <param name="method">任务方法</param>
		/// <param name="argument">任务参数</param>
		/// <returns>任务编号</returns>
		public Int32 Queue(WaitCallback method, Object argument)
		{
			return Queue(new ThreadTask(method, argument));
		}

		/// <summary>把用户工作项放入队列。指定任务被取消时执行的方法，该方法仅针对尚未被线程开始调用时的任务有效</summary>
		/// <param name="method">任务方法</param>
		/// <param name="abortMethod">任务被取消时执行的方法</param>
		/// <param name="argument">任务参数</param>
		/// <returns>任务编号</returns>
		public Int32 Queue(WaitCallback method, WaitCallback abortMethod, Object argument)
		{
			return Queue(new ThreadTask(method, abortMethod, argument));
		}

		/// <summary>把用户工作项放入队列</summary>
		/// <param name="task">任务</param>
		/// <returns>任务编号</returns>
		private Int32 Queue(ThreadTask task)
		{
			//加锁，防止冲突
			lock (Sync_Tasks)
			{
				Tasks.Add(task.ID, task);

				//初始化线程池
				if (ManagerThread == null || !ManagerThread.IsAlive)
				{
					Init();
				}
			}

			//通知管理线程，任务到达
			Event.Set();
			return task.ID;
		}

		/// <summary>取消任务</summary>
		/// <param name="id">任务编号</param>
		/// <returns>任务状态</returns>
		public TaskState Abort(Int32 id)
		{
			// 重点：
			// 这里使用了锁，很危险，所以仅仅在锁里面删除任务，任务的善后处理在锁外面完成
			// 要取消的任务
			ThreadTask task = null;

			// 任务状态
			var state = TaskState.Finished;

			#region 检查任务是否还在队列里面

			if (Tasks.ContainsKey(id))
			{
				//加锁，防止冲突
				lock (Sync_Tasks)
				{
					if (Tasks.ContainsKey(id))
					{
						task = Tasks[id];
						Tasks.Remove(id);
						state = TaskState.Unstarted;
					}
				}
			}

			#endregion

			#region 检查任务是否正在处理

			if (task == null && Threads.Count > 0)
			{
				lock (SyncLock_Threads)
				{
					if (Threads.Count > 0)
					{
						foreach (var item in Threads)
						{
							if (item.Task != null && item.Task.ID == id)
							{
								task = item.Task;
								var b = item.Running;
								item.Abort(true);
								if (b)
								{
									state = TaskState.Running;
								}
								else
								{
									state = TaskState.Finished;
								}
							}
						}
					}
				}
			}

			#endregion

			if (task == null)
			{
				state = TaskState.Finished;
			}

			// 处理任务结束时的事情
			if (task != null && task.AbortMethod != null)
			{
				try
				{
					task.AbortMethod(task.Argument);
				}
				catch { }
			}
			return state;
		}

		/// <summary>取消所有未开始任务</summary>
		/// <remarks>这里不要调用上面Abort取消单个任务，否则可能会造成死锁</remarks>
		public void AbortAllTask()
		{
			// 重点：
			// 这里使用了锁，很危险，所以仅仅在锁里面删除任务，任务的善后处理在锁外面完成
			if (Tasks == null || Tasks.Count < 1) { return; }
			List<ThreadTask> list = null;
			lock (Sync_Tasks)
			{
				if (Tasks == null || Tasks.Count < 1) { return; }
				list = new List<ThreadTask>();

				foreach (var item in Tasks.Values)
				{
					list.Add(item);
				}
				Tasks.Clear();
			}
			if (list == null || list.Count < 1) { return; }

			foreach (var item in list)
			{
				if (item.AbortMethod != null)
				{
					try
					{
						item.AbortMethod(item.Argument);
					}
					catch { }
				}
			}
		}

		/// <summary>取消所有进行中任务</summary>
		/// <remarks>这里不要调用上面Abort取消单个任务，否则可能会造成死锁</remarks>
		public void AbortAllThread()
		{
			// 重点：
			// 这里使用了锁，很危险，所以仅仅在锁里面删除任务，任务的善后处理在锁外面完成
			if (Threads == null || Threads.Count < 1) { return; }
			List<ThreadTask> list = null;
			lock (SyncLock_Threads)
			{
				if (Threads == null || Threads.Count < 1) { return; }
				list = new List<ThreadTask>();

				foreach (var item in Threads)
				{
					if (item.Running)
					{
						list.Add(item.Task);
						item.Abort(true);
					}
				}
			}
			if (list == null || list.Count < 1) { return; }

			foreach (var item in list)
			{
				if (item.AbortMethod != null)
				{
					try
					{
						item.AbortMethod(item.Argument);
					}
					catch { }
				}
			}
		}

		/// <summary>取消所有任务</summary>
		/// <remarks>这里不要调用上面Abort取消单个任务，否则可能会造成死锁</remarks>
		public void AbortAll()
		{
			AbortAllTask();
			AbortAllThread();
		}

		/// <summary>查询任务状态</summary>
		/// <param name="id">任务编号</param>
		/// <returns>任务状态</returns>
		public TaskState Query(Int32 id)
		{
			if (Tasks == null || Tasks.Count < 1) { return TaskState.Unstarted; }

			//检查任务是否还在队列里面
			if (Tasks.ContainsKey(id)) { return TaskState.Unstarted; }

			//检查任务是否正在处理
			if (Threads == null || Threads.Count < 1) { return TaskState.Finished; }
			lock (SyncLock_Threads)
			{
				if (Threads == null || Threads.Count < 1) { return TaskState.Finished; }

				foreach (var item in Threads)
				{
					if (item.Task != null && item.Task.ID == id)
					{
						if (item.Running)
						{
							return TaskState.Running;
						}
						else
						{
							return TaskState.Finished;
						}
					}
				}
			}
			return TaskState.Finished;
		}

		/// <summary>查询任务个数</summary>
		/// <returns></returns>
		public Int32 QueryCount()
		{
			lock (Sync_Tasks)
			{
				return Tasks.Count;
			}
		}

		/// <summary>等待所有任务完成，并指定是否在等待之前退出同步域。</summary>
		/// <param name="millisecondsTimeout"></param>
		/// <returns>是否在等待之前退出同步域</returns>
		public Boolean WaitAll(Int32 millisecondsTimeout)
		{
			var sw = Stopwatch.StartNew();
			Int32 Interval = 10;

			while (true)
			{
				if (RunningCount < 1)
				{
					try
					{
						if (Tasks.Count < 1) { break; }
					}
					catch (Exception ex)
					{
						WriteLog("取任务数异常！" + ex.ToString());
					}
				}
				if (sw.ElapsedMilliseconds >= millisecondsTimeout) { return false; }
				Thread.Sleep(Interval);
			}
			return true;
		}

		#endregion

		#region -- 维护 --

		/// <summary>调度包装</summary>
		private void Work()
		{
			while (true)
			{
				try
				{
					//等待事件量，超时1秒
					Event.WaitOne(1000, false);
					Event.Reset();
					lock (SyncLock_Threads)
					{
						#region 线程维护与统计

						Int32 freecount = 0;

						//清理死线程
						for (Int32 i = Threads.Count - 1; i >= 0; i--)
						{
							if (Threads[i] == null)
							{
								Threads.RemoveAt(i);
								HmTrace.WriteException(Name + "线程池的线程对象为空，设计错误！");
							}
							else if (!Threads[i].IsAlive)
							{
								Threads[i].Dispose();
								HmTrace.WriteException(Threads[i].Name + "处于非活动状态，设计错误！");
								Threads.RemoveAt(i);
							}
							else if (!Threads[i].Running)
							{
								freecount++;
							}
						}

						//正在处理任务的线程数
						RunningCount = Threads.Count - freecount;
						WriteLog("总数：" + Threads.Count + "  可用：" + freecount + " 任务数：" + Tasks.Count);
						Int32 count = MinThreads - freecount;

						//保留最小线程数个线程
						if (count > 0)
						{
							for (Int32 i = 0; i < count; i++)
							{
								ThreadX thread = AddThread();
								if (thread != null) { Threads.Add(thread); }
							}
						}
						else if (count < 0)//过多活动线程，清理不活跃的
						{
							for (Int32 i = Threads.Count - 1; i >= 0 && count < 0; i--)
							{
								if (Threads[i].CanRelease)
								{
									Threads[i].Dispose();
									Threads.RemoveAt(i);
									count++;
								}
							}
						}

						#endregion
					}

					//检查任务，分派线程
					if (Tasks.Count > 0)
					{
						lock (Sync_Tasks)
						{
							while (Tasks.Count > 0)
							{
								//借一个线程
								ThreadX thread = Open();
								if (thread == null) { break; }
								WriteLog("借得线程" + thread.Name);

								//拿出一个任务
								Int32 id = Tasks.Keys[0];
								thread.Task = Tasks[id];
								Tasks.RemoveAt(0);

								//处理任务
								thread.Start();
							}
						}
					}
				}
				catch (ThreadInterruptedException ex)
				{
					LastError = ex;
					break;
				}
				catch (ThreadAbortException ex)
				{
					LastError = ex;
					break;
				}
				catch (Exception ex)
				{
					LastError = ex;
					HmTrace.WriteException(ex);
				}
			}

			// 结束所有工作了，回家吧
			AbortAll();
		}

		/// <summary>添加线程。本方法不是线程安全，调用者需要自己维护线程安全</summary>
		/// <returns></returns>
		private ThreadX AddThread()
		{
			//保证活动线程数不超过最大线程数
			if (Threads.Count >= MaxThreads) { return null; }
			var thread = new ThreadX();

			//thread.Name = Name + "线程池" + ThreadCount + "号线程";
			//thread.Name = String.Format("{0}线程池{1,3}号线程", Name, ThreadCount);
			thread.Name = Name + "P" + ThreadCount;
			thread.OnTaskFinished += new EventHandler<EventArgs>(thread_OnTaskFinished);
			ThreadCount++;
			////暂停一下
			//Thread.Sleep(10);
			WriteLog("新建线程：" + thread.Name);
			return thread;
		}

		private void thread_OnTaskFinished(object sender, EventArgs e)
		{
			var thread = sender as ThreadX;

			Close(thread);

			//通知管理线程，任务完成
			Event.Set();
		}

		#endregion

		#region -- 线程调度 --

		/// <summary>借用线程</summary>
		/// <returns></returns>
		private ThreadX Open()
		{
			lock (SyncLock_Threads)
			{
				foreach (var item in Threads)
				{
					if (item != null && item.IsAlive && !item.Running) { return item; }
				}

				//没有空闲线程，加一个
				if (Threads.Count < MaxThreads)
				{
					var thread = AddThread();
					Threads.Add(thread);
					RunningCount++;
					return thread;
				}
				else
				{
					WriteLog("已达到最大线程数！");
				}
			}
			return null;
		}

		/// <summary>归还线程</summary>
		/// <param name="thread"></param>
		private void Close(ThreadX thread)
		{
			if (thread == null) { return; }
			WriteLog("归还线程" + thread.Name);
			RunningCount--;

			//看这个线程是活的还是死的，死的需要清除
			if (!thread.IsAlive)
			{
				if (Threads.Contains(thread))
				{
					lock (SyncLock_Threads)
					{
						if (Threads.Contains(thread))
						{
							Threads.Remove(thread);
							HmTrace.WriteException("归还" + thread.Name + "时发现，线程被关闭了，设计错误！");
						}
					}
				}
				thread.Dispose();
			}
		}

		#endregion

		#region -- 全局线程池助手 --

		/// <summary>带异常处理的线程池任务调度</summary>
		/// <param name="callback"></param>
		public static void QueueUserWorkItem(WaitCallback callback)
		{
			QueueUserWorkItem(callback, null);
		}

		/// <summary>带异常处理的线程池任务调度</summary>
		/// <param name="callback"></param>
		/// <param name="state"></param>
		public static void QueueUserWorkItem(WaitCallback callback, Object state)
		{
			QueueUserWorkItem(callback, state, HmTrace.WriteDebug);
		}

		/// <summary>带异常处理的线程池任务调度，即使不指定异常处理方法，也不允许异常抛出，以免造成应用程序退出</summary>
		/// <param name="callback"></param>
		/// <param name="state"></param>
		/// <param name="errCallback">发生异常时调用的方法</param>
		public static void QueueUserWorkItem(WaitCallback callback, Object state, Action<Exception> errCallback)
		{
			if (callback == null) { return; }

			var cb = new WaitCallback(s =>
			{
				var ss = (Object[])s;
				var wcb = ss[0] as WaitCallback;
				var st = ss[1];
				var onerr = ss[2] as Action<Exception>;

				try
				{
					wcb(st);
				}
				catch (Exception ex)
				{
					if (onerr != null)
					{
						try
						{
							onerr(ex);
						}
						catch { }
					}
				}
			});
			ThreadPool.QueueUserWorkItem(cb, new Object[] { callback, state, errCallback });
		}

		/// <summary>带异常处理的线程池任务调度</summary>
		/// <param name="callback"></param>
		public static void QueueUserWorkItem(Action callback)
		{
			QueueUserWorkItem(callback, ex =>
			{
				HmTrace.WriteDebug(ex);
			});
		}

		/// <summary>带异常处理的线程池任务调度，即使不指定异常处理方法，也不允许异常抛出，以免造成应用程序退出</summary>
		/// <param name="callback"></param>
		/// <param name="errCallback">发生异常时调用的方法</param>
		[DebuggerHidden]
		public static void QueueUserWorkItem(Action callback, Action<Exception> errCallback)
		{
			if (callback == null) { return; }

			var cb = new WaitCallback(s =>
			{
				var ss = (Object[])s;
				var func = ss[0] as Action;
				var onerr = ss[1] as Action<Exception>;

				try
				{
					func();
				}
				catch (Exception ex)
				{
					if (onerr != null)
					{
						try
						{
							onerr(ex);
						}
						catch { }
					}
				}
			});
			ThreadPool.QueueUserWorkItem(cb, new Object[] { callback, errCallback });
		}

		#endregion

		#region -- IDisposable 成员 --

		/// <summary>释放资源</summary>
		public void Dispose()
		{
			Dispose(true);
			GC.SuppressFinalize(this);
		}

		private void Dispose(Boolean disposing)
		{
			WriteLog(Name + "线程池释放资源");
			if (Threads != null && Threads.Count > 0)
			{
				lock (SyncLock_Threads)
				{
					if (Threads != null && Threads.Count > 0)
					{
						for (Int32 i = Threads.Count - 1; i >= 0; i--)
						{
							if (Threads[i] != null)
							{
								Threads[i].Dispose();
							}
							Threads.RemoveAt(i);
						}
					}
				}
			}
			if (ManagerThread != null && ManagerThread.IsAlive)
			{
				ManagerThread.Abort();
			}
			if (_Event != null)
			{
				_Event.Close();
			}
		}

		/// <summary>析构</summary>
		~ThreadPoolX()
		{
			Dispose(false);
		}

		#endregion

		#region -- 辅助函数 --

		private static void WriteLog(String msg)
		{
			if (Debug) { HmTrace.WriteInfo("线程：" + Thread.CurrentThread.Name + " 信息：" + msg); }
		}

		/// <summary>已重载。</summary>
		/// <returns></returns>
		public override String ToString()
		{
			return String.Format("{0}线程池，线程数：{1}，任务数：{2}", Name, Threads.Count, Tasks.Count);
		}

		private static Boolean? _Debug;

		/// <summary>是否调试</summary>
		public static Boolean Debug
		{
			get
			{
				if (_Debug != null) { return _Debug.Value; }
				_Debug = SystemConfigs.GetConfig().IsThreadDebug;
				return _Debug.Value;
			}
			set { _Debug = value; }
		}

		#endregion

		#region -- Resets the thread pool --

		/// <summary>Resets the thread pool.</summary>
		/// <param name="maxWorkingThreads">The max working threads.</param>
		/// <param name="maxCompletionPortThreads">The max completion port threads.</param>
		/// <param name="minWorkingThreads">The min working threads.</param>
		/// <param name="minCompletionPortThreads">The min completion port threads.</param>
		/// <remarks>Code taken from SuperSocket Library
		/// &lt;a href="http://www.supersocket.net/"&gt;Kerry Jiang's SuperSocket Library&lt;/a&gt;
		/// </remarks>
		/// <returns></returns>
		public static Boolean ResetThreadPool(Int32? maxWorkingThreads, Int32? maxCompletionPortThreads, Int32? minWorkingThreads, Int32? minCompletionPortThreads)
		{
			if (maxWorkingThreads.HasValue || maxCompletionPortThreads.HasValue)
			{
				Int32 oldMaxWorkingThreads, oldMaxCompletionPortThreads;

				ThreadPool.GetMaxThreads(out oldMaxWorkingThreads, out oldMaxCompletionPortThreads);

				if (!maxWorkingThreads.HasValue) { maxWorkingThreads = oldMaxWorkingThreads; }

				if (!maxCompletionPortThreads.HasValue) { maxCompletionPortThreads = oldMaxCompletionPortThreads; }

				if (maxWorkingThreads.Value != oldMaxWorkingThreads || maxCompletionPortThreads.Value != oldMaxCompletionPortThreads)
				{
					if (!ThreadPool.SetMaxThreads(maxWorkingThreads.Value, maxCompletionPortThreads.Value))
					{
						return false;
					}
				}
			}

			if (minWorkingThreads.HasValue || minCompletionPortThreads.HasValue)
			{
				Int32 oldMinWorkingThreads, oldMinCompletionPortThreads;

				ThreadPool.GetMinThreads(out oldMinWorkingThreads, out oldMinCompletionPortThreads);

				if (!minWorkingThreads.HasValue) { minWorkingThreads = oldMinWorkingThreads; }

				if (!minCompletionPortThreads.HasValue) { minCompletionPortThreads = oldMinCompletionPortThreads; }

				if (minWorkingThreads.Value != oldMinWorkingThreads || minCompletionPortThreads.Value != oldMinCompletionPortThreads)
				{
					if (!ThreadPool.SetMinThreads(minWorkingThreads.Value, minCompletionPortThreads.Value))
					{
						return false;
					}
				}
			}

			return true;
		}

		#endregion
	}
}